package com.cryann.mqttserver.modle;

import io.netty.channel.Channel;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;

import java.util.Map;
import java.util.Optional;
import java.util.Queue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;

/**
 * @author cRyann
 * @date 2018/12/6 0006 15:34
 * @description:
 */
@Data
@Slf4j
public class MqttMessageStore {

    private volatile static MqttMessageStore messageStore = null;
    private Map<String, Queue<MqttSendMessage>> map = new ConcurrentHashMap<>();

    private MqttMessageStore() {
    }

    public static MqttMessageStore instance() {
        if ( messageStore == null ) {
            messageStore = new MqttMessageStore();
        }
        return messageStore;
    }

    public void addMsg(Channel channel, MqttSendMessage message) {
        if ( messageStore != null ) {
            String msgKey = channel.id().asShortText();
            Queue<MqttSendMessage> queue = map.get(msgKey);
            queue = Optional.ofNullable(queue).orElse(new ConcurrentLinkedQueue<>());
            queue.add(message);
            map.put(msgKey, queue);
            log.info("Add Message [{}] queue Size:[{}]", msgKey, queue.size());
        }
    }

    public void delQueue(String msgKey) {
        if ( msgKey != null ) {
            map.remove(msgKey);
            log.info("del Message [{}]", msgKey);
        }
    }


    public Queue<MqttSendMessage> getMq(String key) {
        if ( key != null ) {
            Queue<MqttSendMessage> queue = map.get(key);
            return queue;
        }
        return null;
    }
}
